package com.lmn.stream.rocketmq;

import com.alibaba.fastjson.JSON;
import com.google.gson.Gson;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.util.MimeTypeUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Map;
import java.util.Random;

@RestController
public class TestController {

    private static final Logger log = LoggerFactory.getLogger(TestController.class);

    @Autowired
    private Source producer;

    @GetMapping("send")
    public String send() {
        MBody body = new MBody();
        body.setId(new Random().nextInt() + "");
        body.setName("zs");
        String tag = "a_tag";
        log.info("发送mq消息,tag:{},msgBody:{}", tag, JSON.toJSONString(body));
        try {
            // RocketMQHeaders a;
            Message<MBody> message = MessageBuilder.withPayload(body)
                    .setHeader(MessageConst.PROPERTY_TAGS, tag)   //消息tag
                    .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                    .setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, "3") // 设置延迟级别为 3秒后消费。
                    .build();
            producer.login().send(message);
            return "发送mq消息成功,tag:{" + tag + "},msgBody:{" + JSON.toJSONString(body) + "}";
        } catch (Exception e) {
            e.printStackTrace();
            return "发送mq消息失败,tag:{" + tag + "},msgBody:{" + JSON.toJSONString(body) + "}" + "异常信息：" + e;
        }
    }

    @StreamListener(MyTopic.LOGIN)
    public RocketMQLocalTransactionState onConsumer(@Payload MBody body, @Headers Map headers) {
        log.info("消息header：" + new Gson().toJson(headers));
        log.info("消息bodyer：" + body.toString());
        return RocketMQLocalTransactionState.COMMIT;
    }
}
